package cn.doitedu.demo2;

import cn.doitedu.demo1.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/11/18
 * @Desc: 学大数据，上多易教育
 *
 * -- 营销策略1：营销规则
 *    针对年龄在 20~28 之间，性别为 male的用户，                         -- 静态画像条件
 *    当他发生一个行为：添加购物车 时，即给他推送一个满200-50的优惠活动信息   -- 触发条件
 *
 * -- 营销策略2：营销规则
 *   针对月平均消费额>200，性别为 male的用户，                     -- 静态画像条件
 *   当他发生3次以上 点赞 行为后                                  -- 实时动态画像条件
 *   等他再发生 下单行为，即给他推送一个 xxxx 优惠活动信息           -- 触发条件
 *
 **/
public class Demo2 {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 读kafka中行为数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("dwd-events")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setGroupId("doit42-rtmk")
                .setClientIdPrefix("doit42-client-")
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(logJson -> JSON.parseObject(logJson, EventBean.class));


        KeyedStream<EventBean, Long> keyedStream = beanStream.keyBy(bean -> bean.getUser_id());


        SingleOutputStreamOperator<String> messages = keyedStream.process(new KeyedProcessFunction<Long, EventBean, String>() {
            // 运算机池
            HashMap<String, RuleCalculatorV2> calculatorPool = new HashMap<>();

            // 规则1的实例对象
            RuleCalculatorV2 calculatorV2Rule1;
            // 规则2的实例对象
            RuleCalculatorV2 calculatorV2Rule2;
            @Override
            public void open(Configuration parameters) throws Exception {

                RuntimeContext runtimeContext = getRuntimeContext();

                /**
                 * 构造运算机实例对象
                 */
                calculatorV2Rule1 = new RuleCalculatorV2_Rule1();
                calculatorV2Rule2 = new RuleCalculatorV2_Rule2();


                /**
                 * 调用规则运算机实例对象的初始化
                 */
                calculatorV2Rule1.init(runtimeContext);
                calculatorV2Rule2.init(runtimeContext);

                /**
                 * 将初始化好的规则运算机实例对象，放入一个HashMap
                 *
                 */
                calculatorPool.put("rule-001",calculatorV2Rule1);
                calculatorPool.put("rule-002",calculatorV2Rule2);

            }

            @Override
            public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context ctx, Collector<String> out) throws Exception {

                for (Map.Entry<String, RuleCalculatorV2> entry : calculatorPool.entrySet()) {
                    RuleCalculatorV2 calculator = entry.getValue();
                    calculator.calculate(eventBean,out);

                }
            }
        });


        messages.print();

        env.execute();


    }
}
